package com.lb.bs.client.processor.mgr;

import com.lb.bs.client.annotation.ProcessorMark;
import com.lb.bs.client.annotation.SingletonMark;
import com.lb.bs.client.factory.SingletonFactory;
import com.lb.bs.client.manager.ScanMgr;
import com.lb.bs.client.processor.ProcessorHandleService;
import com.lb.bs.client.processor.ProcessorService;
import com.lb.bs.client.processor.beans.ProcessorBean;
import com.lb.bs.client.util.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;

/**
 * program: LB-BS-Client
 * author: bsworld.xie
 * create: 2019-12-10 17:47
 * description:
 *
 * 事件处理器
 *
 * 1、对静态事件直接派发，进行处理
 *
 * 2、对动态事件，全部放入queue里，然后由单独的线程轮询queue去处理事件
 */
@SingletonMark
public class ProcessorMgr<T> {
    private static final Logger logger = LoggerFactory.getLogger(ProcessorMgr.class);


    private final ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r,"lb-bs-event-thread");
            t.setDaemon(true);
            return t;
        }
    });

    private final BlockingQueue<ProcessorBean> queue = new ArrayBlockingQueue<>(1000, true);

    private final AtomicBoolean processorSwitch = new AtomicBoolean(false);

    private final ProcessorCore<T> processorCore = SingletonFactory.getInstance(ProcessorCore.class);

    /**
     * @desription: 空构造器，在实例化过程中，初始化processor处理器,事件消费者etc...
     */
    public ProcessorMgr() {
        //初始化事件处理器
        ScanMgr scanMgr = SingletonFactory.getInstance(ScanMgr.class);
        Set<Class<? extends ProcessorService>> processorClazzes = scanMgr.scanProcessor();
        if (CollectionUtils.isEmpty(processorClazzes)) {
            return;
        }
        for (Class<? extends ProcessorService> processorClazz : processorClazzes) {
            ProcessorMark[] processorMarks = processorClazz.getAnnotationsByType(ProcessorMark.class);
            ProcessorMark processorMark = processorMarks[0];
            ProcessorService processorService = SingletonFactory.getInstance(processorClazz);
            Class<? extends ProcessorHandleService>[] filterClasses = processorMark.filterClasses();
            if (CollectionUtils.isNotEmpty(filterClasses)) {
                List<? extends ProcessorHandleService> filterProcessors = Arrays.asList(filterClasses).stream()
                        .distinct()
                        .filter(Objects::nonNull)
                        .map(m -> SingletonFactory.getInstance(m))
                        .collect(Collectors.toList());
                processorCore.registerProcessorFilter(processorMark.type(), filterProcessors);
            }
            processorCore.registerProcessorHandle(processorMark.type(), processorService);
        }
        //初始化事件消费者
        boolean success = processorSwitch.compareAndSet(false, true);
        if (success) {
            asyncProcess();
        }
    }

    /**
     * @param
     * @param bean  实体交互bean
     *              bean.async true:
     *                              则进行异步处理，并且对于处理过程中的异常进行捕获且打印日志后忽略
     *                         false:
     *                              则进行同步处理，并且对于处理过程中的异常直接抛出，阻断主流程
     *
     * @see ProcessorCore
     *
     * @description: 对于实际的处理器进行统一包装，方便管理
     */
    public void route(ProcessorBean<T> bean) {
        boolean async = bean.getAsync();
        boolean proSwitch = processorSwitch.get();
        if (!proSwitch) {
            logger.warn("process(), ProcessorMgr has been shut down or not open, async:{}" ,async);
            return;
        }
        if (async) {
            queue.add(bean);
        } else {
            processorCore.dispatch(bean);
        }
    }




    private void asyncProcess() {
        executorService.execute(() -> {
            while (processorSwitch.get()) {
                ProcessorBean<T> bean = null;
                try {
                    bean = queue.take();
                    if (bean == null || bean.getProcessorType() == null) {
                        continue;
                    }
                    processorCore.dispatch(bean);
                } catch (Throwable e) {
                    logger.error("asyncProcess(),bean:{} error:", bean, e);
                }
            }
        });
    }


    public void close() {
        processorSwitch.compareAndSet(true, false);
    }


}
